package com.css.wc;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;

import java.io.*;
import java.util.Arrays;
import java.util.Map;

/**
 * 读取文件
 */
public class ReadFileSpout extends BaseRichSpout {
    private SpoutOutputCollector collector;
    private BufferedReader bufferedReader;
    /**
     * 初始化方法，类似于构造器只运行一次
     * 一般用来打开数据库连接，网络连接
     * @param map 配置文件
     * @param topologyContext 上下文对象
     * @param spoutOutputCollector 数据输出收集器
     */
    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        try {
             bufferedReader = new BufferedReader(new FileReader(new File("d://words.txt")));
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
        this.collector=spoutOutputCollector;
    }

    /**
     * 下一个数据，tpule是基本单位
     */
    @Override
    public void nextTuple() {
        String line=null;
        try {
            line= bufferedReader.readLine();
            if(line!=null){
                collector.emit(Arrays.asList(line));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("juzi"));
    }
}
